package io.dy;

import io.dy.constant.MqConstant;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.junit.Test;

import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * @author : dmj
 * @date: 2024/8/26 4:24 下午
 * @description:
 */
public class RocketMqTag {

    @Test
    public void tagProducer() throws Exception{
        DefaultMQProducer producer = new DefaultMQProducer("tag-producer-group");
        producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDRESS);
        producer.start();

        Message message = new Message("tagTopic", "dy", "我是tag消息——01".getBytes());
        Message message2 = new Message("tagTopic", "sy", "我是tag消息——02".getBytes());

        producer.send(message);
        producer.send(message2);

        System.out.println("success ....");
        producer.shutdown();
    }


    @Test
    public void tagCProducer() throws Exception{
        DefaultMQProducer producer = new DefaultMQProducer("tagD-producer-group");
        producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDRESS);
        producer.start();

        for (int i = 0; i < 100000; i++) {
            Message message = new Message("tagCTopic", ("我是tag消息" + i).getBytes());
            producer.send(message);
        }

        System.out.println("success ....");
        producer.shutdown();
    }


    @Test
    public void tagConsumer01() throws Exception{
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("tag-consumer-groupA");
        consumer.setNamesrvAddr(MqConstant.NAME_SRV_ADDRESS);
        consumer.subscribe("tagCTopic", "tagC");

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                System.out.println("我是消费者 tagConsumer01 我在消费消息:" + new String(list.get(0).getBody()));
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        new CountDownLatch(1).await();
    }

    @Test
    public void tagConsumer02() throws Exception{
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("tag-consumer-groupB");
        consumer.setNamesrvAddr(MqConstant.NAME_SRV_ADDRESS);
        consumer.subscribe("tagTopic", "dy || sy");

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                System.out.println("我是消费者 tagConsumer02 我在消费消息:" + new String(list.get(0).getBody()));
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        new CountDownLatch(1).await();

    }

    @Test
    public void testStr(){
        String defaultMQPushConsumer = String.format("%s_%s", "DefaultMQPushConsumer", 3);
        System.out.println(defaultMQPushConsumer);
    }
}
